package com.seewo.binlogsql;

import com.alibaba.fastjson.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.seewo.binlogsql.dto.BinlogDTO;
import com.seewo.binlogsql.handler.DeleteHandle;
import com.seewo.binlogsql.handler.InsertHandle;
import com.seewo.binlogsql.handler.TableMapHandle;
import com.seewo.binlogsql.handler.UpdateHandle;
import com.seewo.binlogsql.vo.DbInfoVo;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.BitSet;
import java.util.HashMap;
import java.util.Map;

/**
 * @author linxixin@cvte.com
 * @version 1.0
 * @description
 */
@Slf4j
@Accessors(chain = true)
public class BinlogListenSql {
    static {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            log.error("没找到jdbc类, 无法使用MYSQL类");
        }
    }

    @Getter
    private BinlogParser    binlogParser = new BinlogParser();
    private DbInfoVo        dbInfoVo;
    @Setter
    private Filter          filter       = new Filter() {
    };
    private BinaryLogClient binaryLogClient;

    public BinlogListenSql(DbInfoVo dbInfoVo) {
        this.dbInfoVo = dbInfoVo;
    }

    private String getFirstBinLogName() {
        String url = "jdbc:mysql://" + dbInfoVo.getHost() + ":" + dbInfoVo.getPort() + "/mysql";
        try (Connection conn = DriverManager.getConnection(url, dbInfoVo.getUsername(), dbInfoVo.getPassword()); Statement statement = conn.createStatement()) {
            ResultSet resultSet = statement.executeQuery("show master logs;");
            while (resultSet.next()) {
                return resultSet.getString("Log_name");
            }
        } catch (SQLException e) {
            log.error("获取binlogName失败", e);
        }
        return null;
    }

    private void initBinlogParser() {
        binlogParser.registerHandle(new InsertHandle(filter), EventType.WRITE_ROWS, EventType.EXT_WRITE_ROWS, EventType.PRE_GA_WRITE_ROWS);
        binlogParser.registerHandle(new DeleteHandle(filter), EventType.DELETE_ROWS, EventType.EXT_DELETE_ROWS, EventType.PRE_GA_DELETE_ROWS);
        binlogParser.registerHandle(new UpdateHandle(filter), EventType.UPDATE_ROWS, EventType.EXT_UPDATE_ROWS, EventType.PRE_GA_UPDATE_ROWS);
        binlogParser.registerHandle(new TableMapHandle(dbInfoVo), EventType.TABLE_MAP);
    }

    public BinlogListenSql connectAndListen() {
        initBinlogParser();

        binaryLogClient = new BinaryLogClient(dbInfoVo.getHost(),
                                              dbInfoVo.getPort(),
                                              dbInfoVo.getUsername(),
                                              dbInfoVo.getPassword());
        log.info("监听程序开始启动...");

/*        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        binaryLogClient.setEventDeserializer(eventDeserializer);*/
/*

        HashMap<Long, String> tableMap = new HashMap<>();
        binaryLogClient.registerEventListener(new BinaryLogClient.EventListener() {
            @Override
            public void onEvent(Event event) {
                // binlog事件
                EventData data = event.getData();
                if (data != null) {
                    if (data instanceof TableMapEventData) {
                        TableMapEventData tableMapEventData = (TableMapEventData) data;
                        tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
                    }
                    // update数据
                    if (data instanceof UpdateRowsEventData) {
                        UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                        String tableName = tableMap.get(updateRowsEventData.getTableId());
                        if (filterData(tableName)) {
                            String eventKey = tableName + ".update";
                            BitSet bset=updateRowsEventData.getIncludedColumns();

                            for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                                String msg = JSON.toJSONString(new BinlogDTO(eventKey, row.getValue()));
                                log.info("update data : {}", msg);
                            }
                        }
                    }
                    // insert数据
                    else if (data instanceof WriteRowsEventData) {
                        WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                        String tableName = tableMap.get(writeRowsEventData.getTableId());
                        if (filterData(tableName)) {
                            String eventKey = tableName + ".insert";
                            for (Serializable[] row : writeRowsEventData.getRows()) {
                                String msg = JSON.toJSONString(new BinlogDTO(eventKey, row));
                                log.info("insert data : {}", msg);
                            }
                        }
                    } // delete数据
                    else if (data instanceof DeleteRowsEventData) {
                        DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                        String tableName = tableMap.get(deleteRowsEventData.getTableId());
                        if (filterData(tableName)) {
                            String eventKey = tableName + ".delete";
                            for (Serializable[] row : deleteRowsEventData.getRows()) {
                                String msg = JSON.toJSONString(new BinlogDTO(eventKey, row));
                                log.info("delete data : {}", msg);
                            }
                        }
                    }
                }
            }
        });
*/

        //binaryLogClient.setServerId(1);
        //binaryLogClient.setBinlogFilename(getFirstBinLogName());

        binaryLogClient.registerEventListener(event -> {

            if (!filter.filter(event)) {
                return;
            }
            binlogParser.handle(event);
        });

        /*new Thread(() -> {
            try {
                binaryLogClient.connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();*/
        try {
            binaryLogClient.connect();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return this;
    }

    private static boolean filterData(String tableName){
        log.info("filterData current tableName : {}",tableName);
        if(tableName == null){
            return false;
        }
        return true;
    }

    public void close() {
        try {
            if (binaryLogClient != null) {
                binaryLogClient.disconnect();
                binaryLogClient = null;
            }
        } catch (IOException e) {
            log.error("关闭失败", e);
        }
    }

}
